package com.jindouyun.consumer.client;

import com.jindouyun.consumer.handle.RpcClientHandle;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @className: com.jindouyun.consumer.client.RpcClient
 * @description: RPC客户端
 * @author: tong.li
 * @createTime: 2021/10/14 14:00
 * @copyright: Tongcheng-Elong Holdings Limited
 */
@Component
public class RpcClient implements InitializingBean, DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);

    private  Channel channel;

    private  NioEventLoopGroup group;

    private String host = "127.0.0.1";

    private int port = 9999;

    @Resource
    RpcClientHandle clientHandle;

    private final ExecutorService executorPool = Executors.newFixedThreadPool(2);

    /**
     * 连接服务器端
     */
    public void connect(String host, int port) {
        group = new NioEventLoopGroup();
        try {
            channel = new Bootstrap().group(group).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) {
                            // 添加客户端字符串编解码器
                            channel.pipeline().addLast(new StringDecoder())
                                    .addLast(new StringEncoder())
                                    // 添加RPC客户端处理器
                                    .addLast(clientHandle);
                        }
                    })
                    .connect(host, port).sync().channel();
            logger.info(String.format("------------RPC服务器端[%s:%s]成功------------",host,port));
        } catch (Exception e) {
            logger.error(String.format("连接RPC服务器端[%s:%s]失败",host,port),e);
            this.close();
        }
    }

    @Override
    public void afterPropertiesSet() {
        this.connect(host, port);
    }


    /**
     * 关闭资源
     */
    public void close(){
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
        executorPool.shutdown();
        logger.info("All thread pool has bean shutdowned");
    }


    /**
     * 释放所有资源
     */
    @Override
    public void destroy() {
        this.close();
    }

    public Object send(String msg) throws ExecutionException, InterruptedException {
        clientHandle.setBody(msg);
        return executorPool.submit(clientHandle).get();
    }


}
